/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.common;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import java.util.Collections;

import org.slf4j.Logger;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.datatype.Packable;
import org.sss.server.SssServer;

/**
 * Method set to get information of Map/Reduce task class.
 */
public class Reflection {
  private static Logger logger = SssServer.getLogger(Reflection.class);
  private Reflection() { // prevent a user from instantiating this class.
  }

  private static boolean isPackable(Type type) {
    return (type instanceof Class) && Packable.class.isAssignableFrom((Class<?>)type);
  }

  private static Class<? extends Packable> asPackable(Type type) throws SssException {
    if (!isPackable(type)) {
      throw new SssException("not a Packable class: " + type);
    }
    return ((Class<?>)type).asSubclass(Packable.class);
  }

  private static Method getUniqueMethod(Class<?> klass, String name) throws SssException {
    Method method = null;
    for (Method m: klass.getDeclaredMethods()) {
      if (m.getName().equals(name) && !m.isBridge()) {
        if (method != null) {
          throw new SssException("duplicated cleanup methods defined: " +
                                 klass.getName());
        }
        method = m;
      }
    }
    return method;
  }

  /**
   * Tuple of Packable class.
   */
  public static class TupleType {
    public TupleType(Class<? extends Packable> keyClass,
                     Class<? extends Packable> valueClass) {
      this.keyClass = keyClass;
      this.valueClass = valueClass;
    }
    public final Class<? extends Packable> keyClass;
    public final Class<? extends Packable> valueClass;

    @Override
    public boolean equals(Object o) {
      if (o == this) {
        return true;
      }
      else if (o == null) {
        return false;
      }
      if (!(o instanceof TupleType)) {
        return false;
      }
      else {
        TupleType that = (TupleType)o;
        return (keyClass == that.keyClass) && (valueClass == that.valueClass);
      }
    }
  }

  /**
   * ProcType.
   */
  public static enum ProcType { MAP, REDUCE }

  /**
   * Information of Mapper/Reducer class.
   */
  public static class ProcInfo {
    private ProcType type;
    private List<TupleType> inlets = new ArrayList<TupleType>();
    private List<TupleType> outlets = new ArrayList<TupleType>();
    private Method method = null;
    private Method cleanupMethod = null;

    /**
     *.
     * NOTE: Create only in createProcInfo().
     */
    private ProcInfo() {}

    public ProcType getType() {
      return type;
    }

    public Method getMethod() {
      return method;
    }
    
    public Method getCleanupMethod() {
      return cleanupMethod;
    }

    public List<TupleType> getInputTypes() {
      return Collections.unmodifiableList(inlets);
    }

    public List<TupleType> getOutputTypes() {
      return Collections.unmodifiableList(outlets);
    }

    public boolean isMapper() {
      return type == Reflection.ProcType.MAP;
    }

    public boolean isReducer() {
      return type == Reflection.ProcType.REDUCE;
    }

    public boolean isAvailableAsCombiner() {
      return (type == Reflection.ProcType.REDUCE &&
              inlets.size() == 1 && outlets.size() == 1);
    }
  }

  public static ProcInfo createProcInfo(Class<?> klass) throws SssException {
    final String kname = klass.getName();
    ProcInfo pi = new ProcInfo();

    // procedure type check
    if (Mapper.class.isAssignableFrom(klass)) {
      pi.type = ProcType.MAP;
    }
    else if (Reducer.class.isAssignableFrom(klass)) {
      pi.type = ProcType.REDUCE;
    }
    else {
      throw new SssException("invalid procedure type: " + kname);
    }

    // ensure that the klass has a public default constructor.
    try {
      klass.getConstructor();
    }
    catch (NoSuchMethodException e) {
      throw new SssException(e);
    }

    setMethod(klass, pi);
    setCleanupMethod(klass, pi);

    return pi;
  }

  private static void setMethod(Class<?> klass, ProcInfo pi) throws SssException {
    final String kname = klass.getName();
    final String name = methodName(pi.type);
    final Class<?> ctxKlass = contextClass(pi.type);

    Method method = getUniqueMethod(klass, name);
    if (method == null) {
      throw new SssException("no " + name + " method is defiend: " + kname);
    }

    // map/reduce method's parameter type analysis
    Type[] ts = method.getGenericParameterTypes();
    int i = 0;
    if (ts.length < 1) {
      throw new SssException(name + " has no argument.");
    }

    // Context
    if (!(ts[i] instanceof Class)) {
      throw new SssException("first parameter should be a Context.");
    }
    if (ts[i] != ctxKlass) {
      throw new SssException("first parameter is not an expected one: " +
          ctxKlass.getName() + ", it is " + ts[i]);
    }
    i++;

    // inputs
    while (true) {
      Class<? extends Packable> kclass = null;
      Class<? extends Packable> vclass = null;
      if (i >= ts.length) {
        break;
      }
      if (!isPackable(ts[i])) {
        break;
      }
      kclass = asPackable(ts[i++]);
      if (i >= ts.length) {
        throw new SssException("The key argument exists but the value argument does not exist.");
      }
      switch (pi.type) {
        case MAP:
          vclass = asPackable(ts[i]);
          break;
        case REDUCE:
          if (!(ts[i] instanceof ParameterizedType)) {
            throw new SssException("not an Iterable(reduce value): " + ts[i]);
          }
          ParameterizedType p = (ParameterizedType) ts[i];
          if (p.getRawType() != Iterable.class) {
            throw new SssException("not an Iterable(reduce value): " + ts[i]);
          }
          Type[] atas = p.getActualTypeArguments();
          if (atas.length < 1) {
            throw new SssException("maybe not an Iterable(reduce value): " + ts[i]);
          }
          vclass = asPackable(atas[0]);
          break;
      }
      i++;
      pi.inlets.add(new TupleType(kclass, vclass));
    }

    // outputs
    for (;i < ts.length;++i) {
      pi.outlets.add(getOutputParam(ts[i]));
    }

    if (pi.inlets.isEmpty()) {
      throw new SssException(name + " has no input arguments.");
    }

    if (pi.outlets.isEmpty()) {
      throw new SssException(name + " has no output arguments.");
    }

    pi.method = method;
  }

  private static void setCleanupMethod(Class<?> klass, ProcInfo pi) throws SssException {
    final String kname = klass.getName();
    final Class<?> ctxKlass = contextClass(pi.type);

    Method method = getUniqueMethod(klass, "cleanup");
    if (method == null) {
      return;
    }

    Type[] ts = method.getGenericParameterTypes();
    int i = 0;
    if (ts.length < 1 || ts[i++] != ctxKlass) {
      logger.warn("{} has \"cleanup\" method but first argument is not Context.", kname);
      return;
    }
  
    for (;i < ts.length;++i) {
      try {
        TupleType t = getOutputParam(ts[i]);
        TupleType expect  = pi.outlets.get(i-1);
        if (!t.equals(expect)) {
          logger.warn("mismatch output type");
          return;
        }
      }
      catch (SssException e) {
        logger.warn(e.getMessage());
        return;
      }
    }
    if (i != 1 + pi.outlets.size()) {
      logger.warn("mismatch output type");
      return;
    }
    pi.cleanupMethod = method;
  }

  private static TupleType getOutputParam(Type t) throws SssException {
    if (!(t instanceof ParameterizedType)) {
      throw new SssException("not an Output class: " + t);
    }
    ParameterizedType p = (ParameterizedType) t;
    if (p.getRawType() != Output.class) {
      throw new SssException("not an Output class: " + t);
    }
    Type[] atas = p.getActualTypeArguments();
    if (atas.length != 2) {
      throw new SssException("maybe not an Output<K, V> class: " + t);
    }
    return new TupleType(asPackable(atas[0]), asPackable(atas[1]));
  }

  private static String methodName(ProcType type) {
    switch (type) {
      case MAP:    return "map";
      case REDUCE: return "reduce";
    }
    return null;
  }

  private static Class<?> contextClass(ProcType type) {
    switch (type) {
      case MAP:    return Mapper.Context.class;
      case REDUCE: return Reducer.Context.class;
    }
    return null;
  }
}
